import json
import urllib
from time import sleep
from typing import IO
from urllib.request import Request

from bs4 import BeautifulSoup, Tag
from kafka import KafkaProducer


def get_page(num: int):
    sleep(1)
    url: str = 'https://www.cnhnb.com/p/ningmeng-0-0-0-0-{}/'.format(num)

    user_agent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36 Edg/136.0.0.0"
    headers: dict = {'User-Agent': user_agent,
               'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
               'Cookie': 'deviceId=9c3a1c2-0782-4b0e-bf6a-363a62835; Hm_lvt_81fc10bb72f85b5a9ff93042925f6543=1701049192,1701071960,1702614490; hnUserTicket=d626a00e-4daa-4bc5-95f7-4d65d019e436; hnUserId=249629450; Hm_lvt_0e023fed85d2150e7d419b5b1f2e7c0f=1702550220,1702599669,1703206865,1703208170; Hm_lvt_91cf34f62b9bedb16460ca36cf192f4c=1702550220,1702599660,1703206865,1703208170; fingerId=1b33e742b4ad274d844d887711b8b7d1; sessionId=S_0LQFY67AZCQZX8NS; Hm_lvt_a6458082fb548e5ca7ff77d177d2d88d=1702550221,1702599669,1703206905,1703208176; Hm_lpvt_91cf34f62b9bedb16460ca36cf192f4c=1703208313; Hm_lpvt_0e023fed85d2150e7d419b5b1f2e7c0f=1703208313; Hm_lpvt_a6458082fb548e5ca7ff77d177d2d88d=1703208313'}

    req: Request = Request(url, headers=headers)
    html: IO = urllib.request.urlopen(req)
    soup: BeautifulSoup = BeautifulSoup(html, 'html.parser')
    uci: list[Tag] = soup.find_all('div', class_='show-ctn')
    product_info: list = []

    for i in uci:
        product_name: str = i.find('h2').get_text()  # 产品名称
        supplier: str = i.find("a").get_text()  # 供应商
        # supplier_web = i.find("a").attrs['href']  # 供应商网址
        product_desc: str = i.find('div', class_='shop-image').img.attrs['alt']  # 产品介绍
        # product_img = i.find('div', class_='shop-image').img.attrs['src']  # 产品图片
        # 产品价格
        product_price: str = str(i.find('div', class_='shops-price').get_text()).strip().replace('\n', '').replace(' ', '')
        addr: str = str(i.find('div', class_='r-shop-btm').get_text())  # 发货地
        mailing_ins: str = str(i.find('div', class_='cw-tags').get_text()).strip().replace('\n', '').replace(' ', '')  # 邮寄说明
        product: list[str] = [product_name, supplier, product_desc, product_price, addr, mailing_ins]
        product_info.append(product)

    return product_info


KAFKA_BROKERS: list = [
    '192.168.236.130:9092',
    '192.168.236.131:9092',
    '192.168.236.132:9092'
]
KAFKA_TOPIC: str = 'bda'
KAFKA_GROUP: str = 'bda_group'
TOPIC_NAME: str = 'bda'

producer = KafkaProducer(
    bootstrap_servers=KAFKA_BROKERS,
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    retries=5,
    acks='all',
    linger_ms=1000
)


def send_to_kafka(data) -> None:
    for i in data:
        producer.send(TOPIC_NAME, value=i)
    producer.flush()


if __name__ == '__main__':
    for i in range(1, 11):
        products: list[str] = get_page(i)
        send_to_kafka(products)